package site.teamo.biu.flink.function.process;

import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.googlecode.aviator.AviatorEvaluator;
import com.googlecode.aviator.Expression;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import site.teamo.biu.flink.sdk.annotation.function.BFProcessFunction;
import site.teamo.biu.flink.sdk.function.AbstractProcessFunction;
import site.teamo.biu.flink.sdk.job.JobContext;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

/**
 * @author haocongshun
 * @date 2023/07/27 14:20:47
 */
@BFProcessFunction
public class SimpleFilterFunction extends AbstractProcessFunction<Object, Object> {
    private LogicalRelationship logicalRelationship = LogicalRelationship.OR;
    private List<Expression> predicates;

    public SimpleFilterFunction(String name, JobContext context) {
        super(name, context);

    }

    @Override
    public void open(Configuration parameters) throws Exception {
        this.predicates = new ArrayList<>();
        JSONObject config = context.getConfig(name);
        this.logicalRelationship = Optional.ofNullable(config.getString("logicalRelationship"))
                .map(String::toUpperCase)
                .map(LogicalRelationship::valueOf)
                .orElse(LogicalRelationship.OR);
        JSONArray array = config.getJSONArray("predicates");
        for (int i = 0; i < array.size(); i++) {
            Expression compile = AviatorEvaluator.compile(array.getString(i));
            predicates.add(compile);
        }
    }

    @Override
    public void processElement(Object in, ProcessFunction<Object, Object>.Context context, Collector<Object> collector) throws Exception {
        if (this.logicalRelationship.function.apply(in, this.predicates)) {
            collector.collect(in);
        }
    }

    private static boolean or(Object in, List<Expression> predicates) {
        for (Expression predicate : predicates) {
            Boolean execute = (Boolean) predicate.execute(MapUtil.of("record", in));
            if (execute) {
                return true;
            }
        }
        return false;
    }

    private static boolean and(Object in, List<Expression> predicates) {
        for (Expression predicate : predicates) {
            Boolean execute = (Boolean) predicate.execute(MapUtil.of("record", in));
            if (!execute) {
                return false;
            }
        }
        return true;
    }

    @FunctionalInterface
    private interface LogicalRelationshipFunction {
        boolean apply(Object in, List<Expression> predicates);
    }

    private enum LogicalRelationship {
        OR(SimpleFilterFunction::or),
        AND(SimpleFilterFunction::and);
        public final LogicalRelationshipFunction function;

        LogicalRelationship(LogicalRelationshipFunction function) {
            this.function = function;
        }
    }

}
